package xin.marcher.module.diplomacy.manager.common;

import cn.hutool.core.map.MapUtil;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import xin.marcher.framework.common.util.json.JsonUtils;
import xin.marcher.module.common.redis.RedisBloom;
import xin.marcher.module.common.redis.RedisCache;
import xin.marcher.module.common.redis.RedisLock;
import xin.marcher.module.common.utils.diplomacy.RedisKeyUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Stream;

import static java.util.stream.Collectors.toList;
import static xin.marcher.module.common.constants.diplomacy.BaseConstants.*;

/**
 * 缓存读写管理
 */
@Slf4j
@Service
public class RedisReadWriteManager {

    @Resource
    private RedisCache redisCache;

    @Resource
    private RedisBloom redisBloom;

    @Resource
    private RedisLock redisLock;

    private static final int SORTED_SET_MAX = 0;
    private static final int SORTED_SET_OFFSET = 0;

    /**
     * 批量获取缓存数据
     *
     * @param useLocalCache       是否使用本地缓存
     * @param keyList             关键字列表
     * @param bloomKey            布隆过滤器 key
     * @param getInitBloomValue   布隆过滤器初始化值方法
     * @param bloomField          布隆过滤器初始化值的字段
     * @param clazz               需要将缓存JSON转换的对象
     * @param getRedisKeyFunction 获取 redis key 的方法
     * @param getDbFunction       获取数据源对象的方法
     * @return java.util.Optional<java.util.List < T>>
     */
    public <T> Optional<List<T>> listRedisStringDataByCache(Boolean useLocalCache,
                                                            Collection<String> keyList,
                                                            String bloomKey,
                                                            Function<String, Optional<List<String>>> getInitBloomValue,
                                                            String bloomField,
                                                            Class<T> clazz,
                                                            Function<String, String> getRedisKeyFunction,
                                                            Function<String, Optional<T>> getDbFunction) {
        try {
            List<T> list = Lists.newArrayList();
            List<String> pendingKeyList;
            // 首先过滤掉布隆过滤器中不存在的数据和重复的数据
            Stream<String> distinctKeyStream = keyList.stream().distinct();
            if (StringUtils.isNotBlank(bloomKey)) {
                // TODO redisBloom.contains 暂返回 ture
                pendingKeyList = distinctKeyStream.filter(key -> redisBloom.contains(bloomKey, key, getInitBloomValue, bloomField)).collect(toList());
            } else {
                pendingKeyList = distinctKeyStream.collect(toList());
            }

            List<String> redisKeyList = pendingKeyList.stream().map(getRedisKeyFunction).distinct().collect(toList());
            List<String> cacheList = redisCache.mget(useLocalCache, redisKeyList);
            for (int i = 0; i < cacheList.size(); i++) {
                String cache = cacheList.get(i);
                // 过滤无效缓存, 缓存值为 {} 的情况
                if (EMPTY_OBJECT_STRING.equals(cache)) {
                    continue;
                }
                // 缓存数据存在
                if (StringUtils.isNotBlank(cache)) {
                    if (JSON.isValidArray(cache)) {
                        List<T> tList = JsonUtils.parseArray(cache, clazz);
                        list.addAll(tList);
                    } else {
                        T t = JsonUtils.parseObject(cache, clazz);
                        list.add(t);
                    }
                    continue;
                }

                // 缓存没有则读库
                Optional<T> optional = getRedisStringDataByDb(useLocalCache, pendingKeyList.get(i),
                        getRedisKeyFunction, getDbFunction);
                // 添加元素至 list
                optional.ifPresent(list::add);
            }

            return CollectionUtils.isEmpty(list) ? Optional.empty() : Optional.of(list);
        } catch (Exception e) {
            log.error("批量获取缓存数据异常 keyList={},clazz={}", keyList, clazz, e);
            throw e;
        }
    }

    /**
     * 获取缓存数据
     *
     * @param useLocalCache       是否使用本地缓存
     * @param key                 关键字
     * @param bloomKey            布隆过滤器key
     * @param getInitBloomValue   布隆过滤器初始化值方法
     * @param bloomField          布隆过滤器初始化值的字段
     * @param clazz               需要将缓存JSON转换的对象
     * @param getRedisKeyFunction 获取redis key的方法
     * @return
     */
    public <T> Optional<T> getRedisStringDataByCache(Boolean useLocalCache,
                                                     String key,
                                                     String bloomKey,
                                                     Function<String, Optional<List<String>>> getInitBloomValue,
                                                     String bloomField,
                                                     Class<T> clazz,
                                                     Function<String, String> getRedisKeyFunction) {
        try {
            if (!redisBloom.contains(bloomKey, key, getInitBloomValue, bloomField)) {
                return Optional.empty();
            }

            String redisKey = getRedisKeyFunction.apply(key);
            String cacheString = redisCache.get(useLocalCache, redisKey);

            if (EMPTY_OBJECT_STRING.equals(cacheString)) {
                return Optional.empty();
            }
            if (StringUtils.isNotBlank(cacheString)) {
                return Optional.of(JSON.parseObject(cacheString, clazz));
            }
            return Optional.empty();
        } catch (Exception e) {
            log.error("获取缓存数据异常 key={},clazz={}", key, clazz, e);
            throw e;
        }
    }

    /**
     * 批量获取多缓存数据
     *
     * @param keyList
     * @param clazz
     * @param getRedisKeyFunction
     * @param getDbFuction
     * @return java.util.Optional<java.util.List < T>>
     */
    public <T> Optional<List<T>> listRedisStringDataByBatchCache(List<String> keyList, Class<T> clazz, Function<String, String> getRedisKeyFunction, Function<String, Optional<List<T>>> getDbFuction) {
        try {
            List<T> list = Lists.newArrayList();
            List<String> pendingKeyList = keyList.stream().distinct().collect(toList());
            List<String> redisKeyList = pendingKeyList.stream().map(getRedisKeyFunction).distinct().collect(toList());
            List<String> cacheList = redisCache.mget(redisKeyList);
            for (int i = 0; i < cacheList.size(); i++) {
                String cache = cacheList.get(i);
                // 过滤无效缓存
                if (EMPTY_OBJECT_STRING.equals(cache)) {
                    continue;
                }
                if (StringUtils.isNotBlank(cache)) {
                    List<T> tList = JSON.parseArray(cache, clazz);
                    list.addAll(tList);
                    continue;
                }
                // 缓存没有则读库
                Optional<List<T>> optional = listRedisStringDataByDb(pendingKeyList.get(i), getRedisKeyFunction, getDbFuction);
                if (optional.isPresent()) {
                    list.addAll(optional.get());
                }
            }

            if (cacheList.isEmpty()) {
                for (int i = 0; i < pendingKeyList.size(); i++) {
                    // 缓存没有则读库
                    Optional<List<T>> optional = listRedisStringDataByDb(pendingKeyList.get(i), getRedisKeyFunction, getDbFuction);
                    if (optional.isPresent()) {
                        list.addAll(optional.get());
                    }
                }
            }

            return CollectionUtils.isEmpty(list) ? Optional.empty() : Optional.of(list);
        } catch (Exception e) {
            log.error("批量获取缓存数据异常 keyList={},clazz={}", keyList, clazz, e);
            throw e;
        }
    }

    /**
     * 读取数据库表数据赋值到redis
     *
     * @param key
     * @param getRedisKeyFunction
     * @param getDbFunction
     * @return java.util.Optional<T>
     */
    public <T> Optional<T> getRedisStringDataByDb(Boolean useLocalCache, String key,
                                                  Function<String, String> getRedisKeyFunction, Function<String, Optional<T>> getDbFunction) {
        if (StringUtils.isEmpty(key) || Objects.isNull(getDbFunction)) {
            return Optional.empty();
        }

        if (!redisLock.lock(key)) {
            return Optional.empty();
        }
        try {
            String redisKey = getRedisKeyFunction.apply(key);
            Optional<T> optional = getDbFunction.apply(key);
            if (!optional.isPresent()) {
                // 把空对象暂存到redis
                redisCache.setex(useLocalCache, redisKey, EMPTY_OBJECT_STRING,
                        RedisKeyUtils.redisKeyRandomTime(INT_EXPIRED_ONE_DAY, TimeUnit.HOURS, NUMBER_24));
                log.warn("发生缓存穿透 redisKey={}", redisKey);
                return optional;
            }
            // 把表数据对象存到redis
            redisCache.setex(useLocalCache, redisKey, JsonUtils.toJsonString(optional.get()), RedisKeyUtils.redisKeyRandomTime(INT_EXPIRED_SEVEN_DAYS));
            log.info("表数据对象存到redis redisKey={}, data={}", redisKey, optional.get());
            return optional;
        } finally {
            redisLock.unlock(key);
        }
    }

    /**
     * 读取数据库表多数据赋值到redis
     *
     * @param key
     * @param getRedisKeyFunction
     * @param getDbFuction
     * @return java.util.Optional<java.util.List < T>>
     */
    public <T> Optional<List<T>> listRedisStringDataByDb(String key, Function<String, String> getRedisKeyFunction, Function<String, Optional<List<T>>> getDbFuction) {
        if (StringUtils.isEmpty(key) || Objects.isNull(getDbFuction)) {
            return Optional.empty();
        }

        try {
            if (!redisLock.lock(key)) {
                return Optional.empty();
            }
            String redisKey = getRedisKeyFunction.apply(key);
            Optional<List<T>> optional = getDbFuction.apply(key);
            if (!optional.isPresent()) {
                // 把空对象暂存到redis
                redisCache.setex(true, redisKey, EMPTY_OBJECT_STRING,
                        RedisKeyUtils.redisKeyRandomTime(INT_EXPIRED_ONE_DAY, TimeUnit.HOURS, NUMBER_24));
                log.warn("发生缓存穿透 redisKey={}", redisKey);
                return optional;
            }
            // 把表数据对象存到redis
            redisCache.setex(true, redisKey, JSON.toJSONString(optional.get()), RedisKeyUtils.redisKeyRandomTime(INT_EXPIRED_SEVEN_DAYS));
            log.info("表数据对象存到redis redisKey={}, data={}", redisKey, optional.get());
            return optional;
        } finally {
            redisLock.unlock(key);
        }
    }

    /**
     * 批量添加string缓存
     *
     * @param redisMap
     * @return void
     */
    public <T> void batchWriteRedisString(Map<String, T> redisMap) {
        List<Map.Entry<String, T>> list = Lists.newArrayList(redisMap.entrySet());
        try {
            for (List<Map.Entry<String, T>> entries : Lists.partition(list, PAGE_SIZE_100)) {
                for (Map.Entry<String, T> entry : entries) {
                    redisCache.setex(true, entry.getKey(), JSON.toJSONString(entry.getValue()),
                            RedisKeyUtils.redisKeyRandomTime(INT_EXPIRED_SEVEN_DAYS));
                }
                try {
                    Thread.sleep(SLEEP_100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        } catch (Exception e) {
            log.error("批量添加string缓存异常 redisMap={}", redisMap, e);
        }
    }

    /**
     * 判断set是否存在
     *
     * @param inputMap
     * @param getRedisKeyFunction
     * @return java.util.Optional<java.util.Map < java.lang.String, java.lang.Boolean>>
     */
    public Optional<Map<String, Boolean>> isExistSet(Map<String, String> inputMap, Function<String, String> getRedisKeyFunction) {
        if (MapUtil.isEmpty(inputMap)) {
            return Optional.empty();
        }
        Map<String, Boolean> resultMap = Maps.newHashMap();
        for (Map.Entry<String, String> entry : inputMap.entrySet()) {
            String key = entry.getKey();
            String redisKey = getRedisKeyFunction.apply(key);
            String member = entry.getValue();
            resultMap.put(key, redisCache.sismember(redisKey, member));
        }
        return Optional.of(resultMap);
    }

    /**
     * 批量判断set是否存在
     *
     * @param key
     * @param memberList
     * @param getRedisKeyFunction
     * @return java.util.Optional<java.util.Map < java.lang.String, java.lang.Boolean>>
     */
    public Optional<Map<String, Boolean>> batchIsExistSet(String key, List<String> memberList, Function<String, String> getRedisKeyFunction) {
        if (CollectionUtils.isEmpty(memberList)) {
            return Optional.empty();
        }
        Map<String, Boolean> resultMap = Maps.newHashMap();
        String redisKey = getRedisKeyFunction.apply(key);
        for (String member : memberList) {
            resultMap.put(member, redisCache.sismember(redisKey, member));
        }

        return Optional.of(resultMap);
    }

    /**
     * 批量获取sorted set
     *
     * @param key
     * @param offset
     * @param count
     * @param getRedisKeyFunction
     * @return java.util.Optional<java.util.List < java.lang.String>>
     */
    public <T> Optional<List<String>> getRedisSortedSet(T key, long offset, long count,
                                                        Function<T, String> getRedisKeyFunction) {
        if (Objects.isNull(key)) {
            return Optional.empty();
        }

        String redisKey = getRedisKeyFunction.apply(key);
        Set<String> strings = redisCache.zrangeByScore(redisKey, 0L, Long.MAX_VALUE, offset, count);

        return CollectionUtils.isEmpty(strings) ? Optional.empty() : Optional.of(new LinkedList<>(strings));
    }


    /**
     * 刷新缓存—差异数据
     *
     * @param dataMap
     * @param getRedisKeyFunction
     * @return void
     */
    public void flushRedisSetDifferenceData(Map<String, Set<String>> dataMap, Function<String, String> getRedisKeyFunction) {
        if (MapUtil.isEmpty(dataMap)) {
            return;
        }

        // 从redis获取数据
        Optional<Map<String, Set<String>>> redisMapOpt = getRedisSetMap(dataMap.keySet(), getRedisKeyFunction);
        Map<String, Set<String>> redisMap = redisMapOpt.orElseGet(HashMap::new);

        // 缓存刷新
        for (Map.Entry<String, Set<String>> entry : dataMap.entrySet()) {
            String key = entry.getKey();
            Set<String> dbSet = entry.getValue();
            Set<String> redisSet = redisMap.getOrDefault(key, Sets.newHashSet());
            Set<String> dbSetResult = Sets.difference(dbSet, redisSet);
            if (!CollectionUtils.isEmpty(dbSetResult)) {
                addSet(getRedisKeyFunction.apply(key), dbSetResult);
            }
            Set<String> redisSetResult = Sets.difference(redisSet, dbSet);
            if (!CollectionUtils.isEmpty(redisSetResult)) {
                removeSet(getRedisKeyFunction.apply(key), redisSetResult);
            }
        }
    }

    /**
     * 刷新缓存—指定的sorted set
     *
     * @param dataMap
     * @param getRedisKeyFunction
     * @return void
     */
    public void flushAssignSortedSetData(Map<String, Map<String, Double>> dataMap, Function<String, String> getRedisKeyFunction) {
        if (MapUtil.isEmpty(dataMap)) {
            return;
        }

        // 保存sorted set
        addSortedSet(dataMap, getRedisKeyFunction);

        // 删除sorted set
        Map<String, Set<String>> deleteMap = Maps.newHashMap();
        for (Map.Entry<String, Map<String, Double>> entry : dataMap.entrySet()) {
            String key = entry.getKey();
            Map<String, Double> memberMap = entry.getValue();

            // 从redis获取数据
            String redisKey = getRedisKeyFunction.apply(key);
            Set<String> oldKeySet = redisCache.zrevrangeByScore(redisKey, Long.MAX_VALUE, SORTED_SET_MAX, SORTED_SET_OFFSET, Integer.MAX_VALUE);
            Set<String> newKeySet = memberMap.keySet();
            Set<String> deleteSet = Sets.difference(oldKeySet, newKeySet);
            if (!CollectionUtils.isEmpty(deleteSet)) {
                deleteMap.put(key, deleteSet);
            }
        }
        if (MapUtil.isNotEmpty(deleteMap)) {
            removeSortedSet(deleteMap, getRedisKeyFunction);
        }
    }

    /**
     * 批量获取set类型数据
     *
     * @param keySet
     * @param getRedisKeyFunction
     * @return java.util.Optional<java.util.Map < java.lang.String, java.util.Set < java.lang.String>>>
     */
    public Optional<Map<String, Set<String>>> getRedisSetMap(Set<String> keySet, Function<String, String> getRedisKeyFunction) {
        Map<String, Set<String>> responseMap = Maps.newHashMap();

        for (String key : keySet) {
            String redisKey = getRedisKeyFunction.apply(key);
            Set<String> valueSet = redisCache.smembers(redisKey);
            if (CollectionUtils.isEmpty(valueSet)) {
                continue;
            }
            responseMap.put(key, valueSet);
        }

        return Optional.of(responseMap);
    }

    /**
     * 添加set
     *
     * @param redisKey
     * @param set
     * @return void
     */
    public void addSet(String redisKey, Set<String> set) {
        redisCache.sadd(redisKey, set.toArray(new String[]{}));
    }

    /**
     * 删除set
     *
     * @param redisKey
     * @param set
     * @return void
     */
    public void removeSet(String redisKey, Set<String> set) {
        redisCache.srem(redisKey, set.toArray(new String[]{}));
    }

    /**
     * 重置set
     *
     * @param redisKey
     * @param set
     * @return void
     */
    public void resetSet(String redisKey, Set<String> set) {
        redisCache.delete(redisKey);
        redisCache.sadd(redisKey, set.toArray(new String[]{}));
    }

    /**
     * 添加sorted set
     *
     * @param addMap
     * @param getRedisKeyFunction
     * @return void
     */
    public void addSortedSet(Map<String, Map<String, Double>> addMap, Function<String, String> getRedisKeyFunction) {
        if (MapUtil.isEmpty(addMap)) {
            return;
        }

        for (Map.Entry<String, Map<String, Double>> entry : addMap.entrySet()) {
            String key = entry.getKey();
            Map<String, Double> memberMap = entry.getValue();
            if (MapUtil.isEmpty(memberMap)) {
                continue;
            }
            String redisKey = getRedisKeyFunction.apply(key);
            for (Map.Entry<String, Double> childEntry : memberMap.entrySet()) {
                String member = childEntry.getKey();
                Double score = childEntry.getValue();
                redisCache.zadd(redisKey, member, score);
            }
        }
    }

    /**
     * 删除sorted set
     *
     * @param removeMap
     * @param getRedisKeyFunction
     * @return void
     */
    public void removeSortedSet(Map<String, Set<String>> removeMap, Function<String, String> getRedisKeyFunction) {
        if (MapUtil.isEmpty(removeMap)) {
            return;
        }

        for (Map.Entry<String, Set<String>> entry : removeMap.entrySet()) {
            String key = entry.getKey();
            Set<String> memberSet = entry.getValue();
            if (CollectionUtils.isEmpty(memberSet)) {
                continue;
            }
            String redisKey = getRedisKeyFunction.apply(key);
            redisCache.zremove(redisKey, memberSet.toArray(new String[]{}));
        }
    }

    /**
     * 删除指定的key
     *
     * @param keys
     * @return void
     */
    public void delete(String... keys) {
        Arrays.asList(keys).stream().forEach(key -> redisCache.delete(key));
    }

    /**
     * 添加sorted set
     *
     * @param redisKey
     * @param addMap
     * @return void
     */
    public void addSortedSet(String redisKey, Map<String, Double> addMap) {
        if (MapUtil.isEmpty(addMap)) {
            return;
        }

        for (Map.Entry<String, Double> entry : addMap.entrySet()) {
            String member = entry.getKey();
            Double score = entry.getValue();
            redisCache.zadd(redisKey, member, score);
        }
    }

    /**
     * 删除sorted set
     *
     * @param redisKey
     * @param memberSet
     * @return void
     */
    public void removeSortedSet(String redisKey, Set<String> memberSet) {
        redisCache.zremove(redisKey, memberSet.toArray(new String[]{}));
    }

    /**
     * 有序成员总数
     *
     * @param redisKey
     * @return java.lang.Long
     */
    public Long zcard(String redisKey) {
        return redisCache.zcard(redisKey);
    }

    /**
     * 有序区间值
     *
     * @param key
     * @param min
     * @param max
     * @return java.util.Set
     */
    public Set zrange(String key, double min, double max) {
        return redisCache.zrangeByScore2(key, min, max);
    }

    /**
     * 批量获取有序列表结果
     *
     * @param keys
     * @param getRedisKeyFunction
     * @param <T>
     * @return
     */
    public <T> Optional<Map<T, List<String>>> queryRedisSortedSetDataMap(List<T> keys,
                                                                         Function<T, String> getRedisKeyFunction) {


        if (CollectionUtils.isEmpty(keys)) {
            return Optional.empty();
        }
        Map<T, List<String>> resultMap = new HashMap<>(keys.size());
        for (T key : keys) {
            Optional<List<String>> optional = getRedisSortedSet(key, 0, -1, getRedisKeyFunction);
            resultMap.put(key, optional.orElseGet(ArrayList::new));
        }

        return Optional.of(resultMap);
    }

    /**
     * 添加新数据至布隆过滤器
     *
     * @param filterKey
     * @param value
     * @param getInitBloomValue
     * @param bloomField
     */
    public void addBloomFilter(String filterKey, Object value,
                               Function<String, Optional<List<String>>> getInitBloomValue, String bloomField) {
        redisBloom.add(filterKey, value, getInitBloomValue, bloomField);
    }
}